package yuekao7.machine;

import com.alibaba.alink.operator.batch.BatchOperator;
import com.alibaba.alink.operator.batch.dataproc.SplitBatchOp;
import com.alibaba.alink.operator.batch.evaluation.EvalMultiClassBatchOp;
import com.alibaba.alink.operator.batch.evaluation.EvalRegressionBatchOp;
import com.alibaba.alink.operator.batch.source.CsvSourceBatchOp;
import com.alibaba.alink.operator.common.evaluation.MultiClassMetrics;
import com.alibaba.alink.operator.common.evaluation.RegressionMetrics;
import com.alibaba.alink.operator.common.evaluation.TuningMultiClassMetric;
import com.alibaba.alink.pipeline.PipelineModel;
import com.alibaba.alink.pipeline.classification.DecisionTreeClassifier;
import com.alibaba.alink.pipeline.classification.LogisticRegression;
import com.alibaba.alink.pipeline.regression.LassoRegression;
import com.alibaba.alink.pipeline.regression.LinearRegression;
import com.alibaba.alink.pipeline.tuning.GridSearchCV;
import com.alibaba.alink.pipeline.tuning.MultiClassClassificationTuningEvaluator;
import com.alibaba.alink.pipeline.tuning.ParamGrid;

public class Boston_Housing {
    public static void main(String[] args) throws Exception {
        //（1）使用Flink批处理加载附件中的boston_housing.csv，封装数据集DataSet，提取特征 features 和标签label，前10条样本数据打印控制台；（3分）
        String filePath = "data/yk7/boston_housing.csv";
        String schema
                //CRIM,ZN,INDUS,CHAS,NOX,RM,AGE,DIS,RAD,TAX,PTRATIO,B,LSTAT,TARGET
                = "CRIM double, ZN double, INDUS double, CHAS double, NOX double, RM double, AGE double, DIS double, RAD double, TAX double" +
                ", PTRATIO double, B double, LSTAT double, TARGET double";
        CsvSourceBatchOp csvSource = new CsvSourceBatchOp()
                .setFilePath(filePath)
                .setSchemaStr(schema)
                .setFieldDelimiter(",");


        String[] features = new String[]{"CRIM", "ZN", "INDUS", "CHAS", "NOX", "RM", "AGE", "DIS", "RAD", "TAX", "PTRATIO", "B", "LSTAT"};
        String label = "TARGET";

//        csvSource.print(10);
        //（2）查看数据的详细信息,并统计数据量，自定义划分训练集和测试集（3分）
//        csvSource.print();
        System.out.println("统计数据量:"+csvSource.print().count());

        BatchOperator<?> spliter = new SplitBatchOp().setFraction(0.8);
        BatchOperator<?> trainData = spliter.linkFrom(csvSource);
        BatchOperator<?> testData = spliter.getSideOutput(0);

        //（3）自主选择使用Alink中至少两种算法构建模型，并设定初始值。（7分）
        //lr
        LinearRegression lr = new LinearRegression()
                .setFeatureCols(features)
                .setLabelCol(label)
                .setPredictionCol("pred")
                .enableLazyPrintModelInfo();
        BatchOperator<?> lr_transform = lr.fit(trainData).transform(testData);
        //lasso
        LassoRegression lasso = new LassoRegression()
                .setFeatureCols(features)
                .setLambda(0.1)
                .setLabelCol(label)
                .setPredictionCol("pred")
                .setMaxIter(1)
                .enableLazyPrintModelInfo();
        BatchOperator<?> lasso_transform = lasso.fit(trainData).transform(testData);
        //（4）对以上算法参数进行调优，可以打印中间调优结果（9分）
        MultiClassClassificationTuningEvaluator multiClassClassificationTuningEvaluator = new MultiClassClassificationTuningEvaluator()
                .setPredictionCol("pred")
                .setLabelCol(label)
                .setTuningMultiClassMetric(TuningMultiClassMetric.ACCURACY);

        ParamGrid paramGrid1 = new ParamGrid()
                .addGrid(lr, LinearRegression.MAX_ITER, new Integer[] {1, 2, 3})
                .addGrid(lr, LinearRegression.NUM_THREADS, new Integer[] {3, 6, 9});

        GridSearchCV cv1 = new GridSearchCV()
                .setEstimator(lr)
                .setParamGrid(paramGrid1)
                .setTuningEvaluator(multiClassClassificationTuningEvaluator)
                .setNumFolds(2)
                .enableLazyPrintTrainInfo("TrainInfo");

        ParamGrid paramGrid2 = new ParamGrid()
                .addGrid(lasso, LassoRegression.MAX_ITER, new Integer[] {1, 2, 3})
                .addGrid(lasso, LassoRegression.NUM_THREADS, new Integer[] {3, 6, 9});

        GridSearchCV cv2 = new GridSearchCV()
                .setEstimator(lasso)
                .setParamGrid(paramGrid2)
                .setTuningEvaluator(multiClassClassificationTuningEvaluator)
                .setNumFolds(2)
                .enableLazyPrintTrainInfo("TrainInfo");

        PipelineModel lr_bestPipelineModel = cv1.fit(trainData).getBestPipelineModel();
        BatchOperator<?> lr_transform1 = lr_bestPipelineModel.transform(testData);

        PipelineModel lasso_bestPipelineModel = cv2.fit(trainData).getBestPipelineModel();
        BatchOperator<?> lasso_transform1 = lasso_bestPipelineModel.transform(testData);
        //（5）自主选择Alink至少两种指标进行模型评估（6分）
        RegressionMetrics metrics1 = new EvalRegressionBatchOp()
                .setPredictionCol("pred")
                .setLabelCol(label)
                .linkFrom(lr_transform)
                .collectMetrics();
        System.out.println("lr Total Samples Number:" + metrics1.getCount());
        System.out.println("lr SSE:" + metrics1.getSse());
        System.out.println("lr SAE:" + metrics1.getSae());
        System.out.println("lr RMSE:" + metrics1.getRmse());
        System.out.println("lr R2:" + metrics1.getR2());


        RegressionMetrics metrics2 = new EvalRegressionBatchOp()
                .setPredictionCol("pred")
                .setLabelCol(label)
                .linkFrom(lasso_transform)
                .collectMetrics();
        System.out.println("lasso Total Samples Number:" + metrics2.getCount());
        System.out.println("lasso SSE:" + metrics2.getSse());
        System.out.println("lasso SAE:" + metrics2.getSae());
        System.out.println("lasso RMSE:" + metrics2.getRmse());
        System.out.println("lasso R2:" + metrics2.getR2());


        MultiClassMetrics metrics3 = new EvalMultiClassBatchOp()
                .setLabelCol(label)
                .setPredictionCol("pred")
                .linkFrom(lr_transform1)
                .collectMetrics();
        System.out.println("lr:"+metrics3.getAccuracy());
        System.out.println("lr:"+metrics3.getMacroRecall());

        MultiClassMetrics metrics4 = new EvalMultiClassBatchOp()
                .setLabelCol(label)
                .setPredictionCol("pred")
                .linkFrom(lasso_transform1)
                .collectMetrics();
        System.out.println("lasso:"+metrics4.getAccuracy());
        System.out.println("lasso:"+metrics4.getMacroRecall());
        //（6）选择较优模型对数据集预测，并打印出结果。（2分）

    }
}
